package com.etlmaster.executor.steps.scheduler;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.quartz.CronScheduleBuilder;
import org.quartz.CronTrigger;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.SchedulerFactory;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import org.quartz.impl.StdSchedulerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.ResultSetExtractor;

import com.etlmaster.executor.Job;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.StepStack;
import com.etlmaster.executor.scheduler.EmJob;
import com.etlmaster.executor.scheduler.ScheduleStatus;
import com.etlmaster.executor.steps.control.LoopStepInterface;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.jdbc.JdbcTemplate;

public class SchedulerStepMeta extends LoopStepInterface {
	private static SchedulerFactory factory = new StdSchedulerFactory();
	private static Scheduler scheduler = null;

	private JdbcTemplate jdbc;

	private String fetchScheduleInfo;
	private List<String> sqls=new ArrayList<String>();
	public SchedulerStepMeta(String id, String name, String step_code,
			String sessionId, String paramStr) {
		super(id, name, step_code, sessionId, paramStr);
		// TODO Auto-generated constructor stub
	}
	

	@Override
	protected void loadParamsFromConfigTable() {
		// TODO Auto-generated method stub

	}

	@Override
	public int process(StepStack stack, StepMeta curStepMeta, Job job) {
		log("INFO", "开始执行调度步骤：{}", curStepMeta.getName());
		jdbc = new JdbcTemplate();
		try {
			scheduler = factory.getScheduler();
			getData(scheduler, job, curStepMeta);
			if (!scheduler.isStarted())
				scheduler.start();
		} catch (Exception e) {

			log("ERROR","调度异常！{}",e.getMessage());
			e.printStackTrace();
			return StepState.RUNNING_FAILED;
		}
		return StepState.RUNNING_SUCCESSED;

	}

	protected void getData(final Scheduler scheduler, final Job dpJob,
			final StepMeta meta) {
		jdbc.query(this.getFetchScheduleInfo(),
				new ResultSetExtractor<List<Map<String, Object>>>() {
					@Override
					public List<Map<String, Object>> extractData(ResultSet rs)
							throws SQLException, DataAccessException {
						while (rs.next()) {
							String taskid = "";
							String jobName = rs.getString("proc_name");
							String agentId=rs.getString("agent_id");
							String concurrrencyrun=rs.getString("concurrencyrun");
							String invalid_date=rs.getString("invalid_date");
							String cycle_type=rs.getString("cycle_type");
							if(agentId==null||agentId.trim().length()==0)
							{
								agentId="null";
							}
							JobKey jobKey = new JobKey(jobName, agentId);
							int jobStatus = rs.getInt("job_status");
							if (!(rs.getString("taskid") == null
									|| rs.getString("taskid").length() == 0 || rs
									.getString("taskid").equalsIgnoreCase(
											"null")))
								taskid = rs.getString("taskid");
							try {
								if (!scheduler.checkExists(jobKey)
										&& jobStatus != ScheduleStatus.invalid) {
									log("INFO","新增流程{}!",jobKey.getName());

									JobDetail job = JobBuilder
											.newJob(EmJob.class)
											.withIdentity(jobName, agentId)
											.build();
									CronTrigger trigger = TriggerBuilder
											.newTrigger()
											.withIdentity("trigger_" + jobName,
													agentId)
											.withSchedule(
													CronScheduleBuilder.cronSchedule(rs
															.getString("cron_exp")))
											.build();
									job.getJobDataMap().put("job", dpJob);
									job.getJobDataMap().put("meta", meta);
									job.getJobDataMap().put("taskid", taskid);
									job.getJobDataMap().put("agentId", agentId);
									job.getJobDataMap().put("concurrencyRun", concurrrencyrun);
									job.getJobDataMap().put("invalid_date", invalid_date);
									job.getJobDataMap().put("cycle_type", cycle_type);

									scheduler.scheduleJob(job, trigger);
								} else {
									if (jobStatus == ScheduleStatus.updated) {
										log("INFO","更新流程{}!",jobKey.getName());

										CronTrigger trigger = TriggerBuilder
												.newTrigger()
												.withIdentity(
														"trigger_" + jobName,
														agentId)
												.withSchedule(
														CronScheduleBuilder
																.cronSchedule(rs
																		.getString("cron_exp")))
												.build();
										TriggerKey triggerKey = new TriggerKey(
												"trigger_" + jobName, agentId);
										JobDetail job = scheduler.getJobDetail(jobKey);
										job.getJobDataMap().put("agentId", agentId);
										job.getJobDataMap().put("concurrencyRun", concurrrencyrun);

										scheduler.rescheduleJob(triggerKey,
												trigger);
										String updateSql=DataCollectionUtils.getProperties("updateScheduleSql").replace("?", jobName);
										sqls.add(updateSql);

									}else if(jobStatus == ScheduleStatus.invalid){
										String baksql=DataCollectionUtils.getProperties("backUpScheduleSql").replace("?", jobName);
										sqls.add(baksql);
										String deleteSql=DataCollectionUtils.getProperties("deleteScheduleSql").replace("?", jobName);
										sqls.add(deleteSql);
										log("INFO","下线流程{}!",jobKey.getName());
										scheduler.deleteJob(jobKey);
									}
								}
							} catch (SchedulerException e) {
								e.printStackTrace();
								log("INFO","调度异常：{}",e.getMessage());
							}

						}

						return null;
					}
				});
		if(sqls.size()!=0){
			String[] strs=new String[sqls.size()];
			for(int i=0;i<sqls.size();i++){
				strs[i]=(String)sqls.get(i);
			}
			jdbc.batchUpdate(strs);
			sqls.clear();
		}

	}
	public String getFetchScheduleInfo() {
		return this.variableSubstitution(fetchScheduleInfo);
	}
	public void setFetchScheduleInfo(String fetchScheduleInfo) {
		this.fetchScheduleInfo = fetchScheduleInfo;
	}



}
